package com.yjjxt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Hello02Transformation {
  def main(args: Array[String]): Unit = {

    //配置并创建对象
    val sparkContext = new SparkContext((new SparkConf().setMaster("local").setAppName("Hello02Transformation" + System.currentTimeMillis())))

    //开始读取数据
    val lines: RDD[String] = sparkContext.textFile("src/main/resources/user.log")
    //过滤数据(只处理包含Hello的)
    val linesFilter: RDD[String] = lines.filter(_.contains("hello"))
    //对每一行数据随机抽样（true放回有重复，false不放回）(抽取的比例)
    val linesSample: RDD[String] = linesFilter.sample(true, 0.5)
    //将行拆成Words
    val words: RDD[String] = linesSample.flatMap(_.split(" "))
    //转换成KV结构，后续进行分组
    val pairs: RDD[(String, Int)] = words.map(ele => {
      println("wordsSample.map-----" + ele)
      new Tuple2(ele, 1)
    })
    //开始进行重新分区
    val reduce: RDD[(String, Int)] = pairs.reduceByKey((sum, value) => {
      println(sum + "--" + value)
      sum + value
    })
    //开始对结果进行排序
    val sort: RDD[(String, Int)] = reduce.sortBy(_._2)


    //开始执行一个行动算子（Starting job 1）
    sort.foreach(println)
    //开始执行第二个行动算子（Starting job 2）
    println(sort.count())

    //最后关闭SparkContext
    sparkContext.stop()
  }
}
